package com.alibaba.dubbo.rpc.protocol.http;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

import org.springframework.remoting.httpinvoker.SimpleHttpInvokerRequestExecutor;
import org.springframework.remoting.support.RemoteInvocation;
import org.springframework.remoting.support.RemoteInvocationResult;

import com.alibaba.dubbo.common.io.Bytes;
import com.alibaba.dubbo.common.io.UnsafeByteArrayInputStream;
import com.alibaba.dubbo.common.io.UnsafeByteArrayOutputStream;
import com.alibaba.dubbo.common.serialize.support.hessian.Hessian2ObjectInput;
import com.alibaba.dubbo.common.serialize.support.hessian.Hessian2ObjectOutput;
import com.alibaba.dubbo.common.serialize.support.hessian.Hessian2Serialization;
import com.alibaba.dubbo.rpc.RpcContext;

public class DubboSimpleHttpInvokerRequestExecutor extends
		SimpleHttpInvokerRequestExecutor {

	public DubboSimpleHttpInvokerRequestExecutor(Class<?> serviceType) {
		super();
		this.serviceType = serviceType;
	}

	Class<?> serviceType;
	AtomicLong seq = new AtomicLong(1);
	
	@Override
	protected ByteArrayOutputStream getByteArrayOutputStream(RemoteInvocation invocation) throws IOException {
		UnsafeByteArrayOutputStream stream = new UnsafeByteArrayOutputStream(4096);
		stream.writeInt8(0xda); //-->0
		stream.writeInt8(0xbb); //-->1
		byte  b = Hessian2Serialization.ID;
		b |= 0x40;//two way
		b |= 0x80;//request
		b |= 0x10;//extend dubbo protocol 
		stream.writeInt8(b); //-->2
		stream.writeInt8(0); //-->3
		
		long req_id = seq.addAndGet(1);
		
		stream.writeInt64b(req_id); //-->4-11
		
		stream.writeInt32b(0); //-->12-15, total length, now is zero
		stream.writeInt32b(0); //-->16-19, data length, now is zero(total = data + header)
		
		ByteArrayOutputStream body = new ByteArrayOutputStream(4096);
		Hessian2ObjectOutput serializer = new Hessian2ObjectOutput(body);
		serializer.writeUTF("2.0.0"); //-->dubbo version
		serializer.writeUTF(this.serviceType.getName()); //->interface 
		serializer.writeUTF(null); //-->version
		serializer.writeUTF(invocation.getMethodName()); ///--> method
		
		serializer.writeLong(invocation.getArguments().length); //-->parameters type, -1 mean auto match
		for(Object arg : invocation.getArguments()){
			serializer.writeObject(arg); //--> parameter
		}
		
		serializer.writeObject(null); //-->attachments
		//must flushBuffer to byteArrayStream
		serializer.flushBuffer();
		
		body.writeTo(stream);
		
		ByteArrayOutputStream headers = new ByteArrayOutputStream(1024);
		serializer = new Hessian2ObjectOutput(headers);
		serializer.writeObject(RpcContext.getContext().getRequestHeaders().getHeaders());
		serializer.flushBuffer();
		headers.writeTo(stream);
		
		//rewrite 12-15, total length
		int dataLength = body.size();
		int headerLength = headers.size();
		int total = dataLength + headerLength + 4;
		stream.writeInt32b(12, total);
		stream.writeInt32b(16, dataLength);
		
		int length = stream.size();
		ByteArrayOutputStream baos = new ByteArrayOutputStream(length);
		stream.writeTo(baos);
		return baos;
	}
	
	
	private int read(InputStream in,  byte[] buffer, int offset, int length) throws IOException{
		int off = offset;
		int len = 0;
		int total = length;
		while(total > 0){
			int r = in.read(buffer, off, total);
			if(r <= 0) break;
			off += r;
			total -= r;
			len += r;
		}
		return len;
	}
	
	@Override
	protected RemoteInvocationResult readRemoteInvocationResult(InputStream input, String codebaseUrl)
			throws IOException, ClassNotFoundException {

		int header_size = 16;
		byte []buffer = new byte[20];
		if(read(input, buffer, 0, 16) != 16) throw new IOException();
		
		 
		
		if(buffer[0] != (byte)0xda && buffer[1] != (byte)0xbb ) throw new IOException();
		
		
		byte flag = buffer[2];
		int total = Bytes.bytes2int(buffer, 12);
		 
		int data = total;
		int header = total - data;
		boolean newProtocol = (flag & 0x10) != 0;
		byte [] xdata;
		if(newProtocol){
			//no enough data
			if(read(input, buffer, 16, 4) != 4)  throw new IOException();
			if(buffer.length < 20) throw new IOException();
			data = Bytes.bytes2int(buffer, 16);
			header = total - data - 4;
			xdata = new byte[header + data];
			input.read(xdata, 0, xdata.length);
		}else{
			xdata = new byte[data];
			input.read(xdata, 0, xdata.length);
		}
		
		long req = Bytes.bytes2int(buffer, 4);
		if((flag & 0x20) != 0){
			//hearbeat
			throw new IOException();
		}
		
		Map<String, String> headers = null;
		//parse header
		if(header > 0){
			Hessian2ObjectInput t = new Hessian2ObjectInput(new UnsafeByteArrayInputStream(xdata, data, header));
			try {
				headers = t.readObject(Map.class);
			} catch (Exception e) {
			}
		}
		
		//parse data
		Hessian2ObjectInput t = new Hessian2ObjectInput(new UnsafeByteArrayInputStream(xdata, 0, data));
		int r = 2;
		Object o;
		try{
			r = t.readInt();
			o = null;
			if(r == 1){
				//value
				o = t.readObject();
			}else if(r == 0){
				//exception;
				o = t.readObject();
			}
		}catch(IOException e){
			r = 0;
			o = e;
		}
		RemoteInvocationResult result = r == 0? new RemoteInvocationResult((Throwable)o):new RemoteInvocationResult(o);
		return result;
	}
}
